package com.sanmubird.mywebframework.eventbus.core;


import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.*;

/**
 * 消息处理中心。
 * 和生产者、消费者建立连接
 * 接收、存储、分发、确认、重试消息
 * 涉及 Channel(建立连接)、Routing key(路由规则)、 Bindings(exchange 和 queue 之间的虚拟连接)、Exchange(分发消息)、Queue(消息队列)
 * 此处的实际应用：
 * 1、将发布消息功能拆分出去，给了一个默认实现，也可以根据需要实现自己的发布消息功能
 * 2、此处核心的两个功能，
 * 一、启动消费者线程，
 * 二、Binding 观察者 和 消费者线程之间的关系，分发规则在此处绑定，分发实现由消费者线程完成。
 */
@Component
public class MqBroker {

    private static final Logger LOGGER = LoggerFactory.getLogger(MqBroker.class);

    @Autowired
    private ApplicationContext applicationContext;

    public LinkedList<MqConsumer> mqConsumers;

    private ConcurrentHashMap<String, List<MqObserver>> OBSERVERS_HOLDER;

    public void setMqConsumers(LinkedList<MqConsumer> mqConsumers) {
        this.mqConsumers = mqConsumers;
    }

    public void setOBSERVERS_HOLDER(ConcurrentHashMap<String, List<MqObserver>> OBSERVERS_HOLDER) {
        this.OBSERVERS_HOLDER = OBSERVERS_HOLDER;
    }


    /**
     * 初始化消费者线程、观察者列表
     */
    @PostConstruct
    void initCollection() {
        mqConsumers = new LinkedList<>();
        String[] consumerThreadNameArr = applicationContext.getBeanNamesForType(MqConsumer.class);
        for (String beanName : consumerThreadNameArr) {
            MqConsumer mqConsumer = (MqConsumer) applicationContext.getBean(beanName);
            mqConsumers.add(mqConsumer);
        }

        OBSERVERS_HOLDER = new ConcurrentHashMap<>();
        String[] observerNameArr = applicationContext.getBeanNamesForType(MqObserver.class);
        for (String beanName : observerNameArr) {
            MqObserver mqObserver = (MqObserver) applicationContext.getBean(beanName);
            List<MqObserver> observers = OBSERVERS_HOLDER.computeIfAbsent(mqObserver.getRoutingKey(), k -> new ArrayList<>());
            observers.add(mqObserver);
        }

        doStartConsumerThread();
    }

    /**
     * 获取观察者列表
     */
    public List<MqObserver> getMqObserverListBy(String routingKey) {
        return OBSERVERS_HOLDER.get(routingKey);
    }

    /**
     * 启动消费者线程
     */
    public void doStartConsumerThread() {
        LOGGER.info("Registering threads to listen redis queue");
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("events-%d").build();
        int size = mqConsumers.size();
        ExecutorService fixedThreadPool = new ThreadPoolExecutor(
                size,
                size,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024),
                namedThreadFactory,
                new ThreadPoolExecutor.CallerRunsPolicy()
        );

        for (MqConsumer mqConsumer : mqConsumers) {
            fixedThreadPool.execute(mqConsumer);
        }
        fixedThreadPool.shutdown();
    }


}
